package club.lovety.code.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MQConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

/**
 * <p>
 * 这里写有象视频代码描述
 * </p>
 *
 * @author sunmc
 * @date created in 2018/5/7 15:10
 * @since
 */
@Service
public class RocketmqConsumer {

    private static final Logger log = LoggerFactory.getLogger(MQConsumer.class);


    @PostConstruct
    public void startConsumer() {

        log.info("开启启用消费者");

        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("last_offset_new_consumer");

        defaultMQPushConsumer.setNamesrvAddr("localhost:9876");

        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        try {
            defaultMQPushConsumer.subscribe("mmm", "*");
//            defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {
//                @Override
//                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
//                    System.out.println("=======开始消费=========:  "+ Thread.currentThread().getName());
//                    log.info("==========开始消费--log4j2===========");
//                    context.setAutoCommit(true);
//                    for(MessageExt messageExt : msgs){
//                        System.out.println("输出消费: "+new String(messageExt.getBody()));
//                        log.info("消费的消息: {}",new String(messageExt.getBody()));
//                    }
//                    return ConsumeOrderlyStatus.SUCCESS;
//                }
//            });


            defaultMQPushConsumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
                        System.out.println("=======开始消费=========:  " + Thread.currentThread().getName());
                        log.error("==========开始消费--log4j2===========");
                        context.setAutoCommit(true);
                        for (MessageExt messageExt : msgs) {
                            System.out.println("输出消费: " + new String(messageExt.getBody()));
                            log.error("消费的消息: {}", new String(messageExt.getBody()));
                        }
                        return ConsumeOrderlyStatus.SUCCESS;
                    }
            );
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
